package com.gis507.mapreduce.analysis_index;

import com.gis507.mapreduce.AISBean.AisClassifyBean;
import com.gis507.mapreduce.AISBean.AnalysisIndex;
import com.gis507.mapreduce.Utils.UtilsClass;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class AISInterpolationMapper extends Mapper<LongWritable,Text, Text, AnalysisIndex> {

    private Text outK = new Text();
    private AnalysisIndex outV = new AnalysisIndex();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        String line = value.toString();

        String[] comment = line.split(",");
        String dest = comment[9];

        double lon_d = new Double(comment[11]);
        //String mmsi, String shipType, String nacStatusCN, double draught,
        // double heading, double course, double speed, String dest,
        // long unixTime, double lon_d, double lat_d, int seaRange
        if ("YANTAI".equals(dest) && lon_d >121.357 && lon_d < 121.983) {


            String mmsi = comment[0];
            double heading = new Double(comment[6]);
            double course = new Double(comment[7]);
            double speed = new Double(comment[8]);
            long unixTime = new Long(comment[10]);
            double lat_d = new Double(comment[12]);

            //fengzhuang
            outK.set(mmsi);

            outV.setMmsi(mmsi);
            outV.setHeading(heading);
            outV.setCourse(course);
            outV.setSpeed(speed);
            outV.setUnixTime(unixTime);
            outV.setLon_d(lon_d);
            outV.setLat_d(lat_d);
            //write for Mapper
            context.write(outK, outV);
        }


    }


}
